package de.distMLP;

import java.io.FileInputStream;
import java.io.IOException;
import java.util.Properties;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hama.HamaConfiguration;
import org.apache.hama.bsp.BSPJob;
import org.apache.hama.bsp.FileOutputFormat;
import org.apache.hama.bsp.TextInputFormat;
import org.apache.hama.bsp.TextOutputFormat;

import de.distMLP.evaluation.MLPEvaluator;
import de.distMLP.train.Configuration;

public class Evaluate_MultilayerPerceptron {

	private static final Path TMP_OUTPUT = new Path("/tmp/pi-" + System.currentTimeMillis());

	public static void main(final String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		final HamaConfiguration conf = Evaluate_MultilayerPerceptron.getConfig(args);

		final BSPJob bsp = new BSPJob(conf, Evaluate_MultilayerPerceptron.class);

		bsp.setJobName("Dist_MultilayerPerceptron");
		bsp.setInputPath(new Path(conf.get(Configuration.TRAININGS_DATA_INPUT_PATH)));

		bsp.setBspClass(MLPEvaluator.class);

		bsp.setInputFormat(TextInputFormat.class);
		bsp.setInputKeyClass(LongWritable.class);
		bsp.setInputValueClass(Text.class);

		bsp.setOutputKeyClass(NullWritable.class);
		bsp.setOutputValueClass(Text.class);
		bsp.setOutputFormat(TextOutputFormat.class);
		FileOutputFormat.setOutputPath(bsp, Evaluate_MultilayerPerceptron.TMP_OUTPUT);

		bsp.setNumBspTask(1);

		if (bsp.waitForCompletion(true)) {
			Helper.printOutput(conf, Evaluate_MultilayerPerceptron.TMP_OUTPUT, true);
		}
	}

	private static HamaConfiguration getConfig(final String[] args) throws IOException {
		if (args.length != 1) {
			System.out.println("Usage: jar <path to properties file>");
			System.exit(1);
		}
		final HamaConfiguration conf = new HamaConfiguration();

		final Properties prop = new Properties();
		prop.load(new FileInputStream(args[0]));

		conf.set(Configuration.TRAININGS_DATA_INPUT_PATH, prop.getProperty(Configuration.TRAININGS_DATA_INPUT_PATH));
		conf.set(Configuration.MLP_INPUT_PATH, prop.getProperty(Configuration.MLP_INPUT_PATH));
		if (prop.containsKey(Configuration.DICTIONARY_PATH)) {
			conf.set(Configuration.DICTIONARY_PATH, prop.getProperty(Configuration.DICTIONARY_PATH));
		}
		return conf;
	}
}
